package cn.com.cmbc.service;

import cn.com.cmbc.config.KafkaConfig;
import cn.com.cmbc.dao.KafkaDAO;
import cn.com.cmbc.utils.JSONUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class KafkaService {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaService.class);
    public boolean sendKafkaList(String topic, List<KafkaDAO> kafkaDAOList) {
        try (KafkaProducer<String, String> producer = new KafkaConfig().getKafkaProducer()) {
            kafkaDAOList.forEach(kafkaDAO -> {
                try {
                    String json = JSONUtil.POJOtoJson(kafkaDAO);
                    if (json != null) {
                        ProducerRecord<String, String> record = new ProducerRecord<>(topic, kafkaDAO.getRowkeyId(), json);
                        producer.send(record);
                    } else {
                        LOG.error("Failed to convert KafkaDAO to JSON: {}", kafkaDAO);
                    }
                } catch (Exception e) {
                    LOG.error("Error converting KafkaDAO to JSON: {}", kafkaDAO, e);
                }
            });
            return true;
        } catch (Exception e) {
            LOG.error("Error in sending Kafka messages", e);
            return false;
        }
    }
}
